package com.lry.raft;

import com.lry.client.RpcClient;
import com.lry.client.RpcProxy;
import com.lry.core.Peer;
import com.lry.core.RaftNode;
import com.lry.message.Message;
import com.lry.message.MessageExt;
import com.lry.message.PullMessage;
import com.lry.producer.SendResult;
import com.lry.serialize.KryoSerializer;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Slf4j
public class MqRaftServiceImpl implements MqRaftService {

    private RaftNode raftNode;

    public MqRaftServiceImpl(RaftNode raftNode){
        this.raftNode = raftNode;
    }

    @Override
    public SendResult sendMsg(Message msg) {
        if(null==raftNode.getLeaderId()){
            log.info("暂时还没有选举出leader，raft处于不可用状态，请稍后重试");
            return null;
        }

        //如果自己不是leader，需要转发给leader处理写请求
        if(raftNode.getLeaderId().intValue()!=raftNode.getServerId()){
            Peer leader = raftNode.getServerId2Peer().get(raftNode.getLeaderId());
            RpcClient rpcClient = new RpcClient(leader.getServer().getHost(),leader.getServer().getPort());

            MqRaftService mqRaftService = RpcProxy.getProxy(rpcClient,MqRaftService.class);
            return mqRaftService.sendMsg(msg);
        }


        //自己是leader, 自己落盘，并且同步给其他从节点，有超过一半的节点落盘成功，则返回成功，否则失败
        byte[] replicate = raftNode.replicate(KryoSerializer.serialize(msg));
        return KryoSerializer.deserialize(replicate);
    }


    @Override
    public List<MessageExt> pullMessage(PullMessage pullMessage) {
        if(null==raftNode.getLeaderId()){
            log.info("暂时还没有选举出leader，raft处于不可用状态，请稍后重试");
            return null;
        }

        byte[] pullMsg = raftNode.get(KryoSerializer.serialize(pullMessage));
        return KryoSerializer.deserialize(pullMsg);
    }

}
